﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Messaging;

namespace curbside.Messaging
{
    /// <summary>
    /// Stager消息队列
    /// </summary>
    public class MessageQueueEx<T> : IDisposable
        where T : BaseMessage
    {
        /// <summary>
        /// 内聚MessageQueue对象
        /// </summary>
        MessageQueue msgQueue;
        TimeSpan timeout = new TimeSpan(0, 0, 0, 0, 10);
        static object _loke = new object();

        /// <summary>
        /// Stager消息队列构造函数
        /// </summary>
        /// <param name="queuePath">消息队列路径</param>
        internal MessageQueueEx(string queuePath)
        {
            msgQueue = new MessageQueue(queuePath);
            msgQueue.DefaultPropertiesToSend.Recoverable = true;
            msgQueue.MessageReadPropertyFilter.SetAll();
            msgQueue.Formatter = new System.Messaging.XmlMessageFormatter(new Type[] { typeof(T) });
        }

        #region 发送消息

        /// <summary>
        /// 发送消息到消息队列
        /// </summary>
        /// <param name="obj">T 消息</param>
        public void Send(T obj)
        {
            if (obj == null)
            {
                throw new ArgumentNullException();
            }
            Message msg = new Message(obj);
            msg.Recoverable = true;
            msg.CorrelationId = GetCorrelationId();
            obj.Id = msg.CorrelationId;
            msgQueue.Send(msg);
        }

        #endregion

        #region 接收消息(接收后删除消息)

        /// <summary>
        /// 引用的队列中可用的第一条消息。此调用是同步的，在有可用消息前，它将一直阻止当前线程的执行。
        /// </summary>
        /// <returns></returns>
        public T Receive()
        {
            return TransferMessageToObject(msgQueue.Receive());
        }

        /// <summary>
        /// 引用的队列中的第一条可用消息，此方法不阻止当前线程的执行。
        /// </summary>
        /// <returns></returns>
        private T ReceiveNoBlock()
        {
            try
            {
                return TransferMessageToObject(msgQueue.Receive(timeout));
            }
            catch (Exception e)
            {
                DealWithException(e);
            }

            return null;
        }

        /// <summary>
        /// 从非事务性队列中接收与给定的标识符匹配的消息
        /// </summary>
        /// <param name="id">消息id</param>
        /// <returns></returns>
        public T ReceiveById(string id)
        {
            try
            {
                return TransferMessageToObject(msgQueue.ReceiveByCorrelationId(id));
            }
            catch (Exception e)
            {
                DealWithException(e);
            }

            return null;
        }

        /// <summary>
        /// 从非事务性队列接收与给定的标识符匹配的消息，此方法不阻止当前线程的执行。
        /// </summary>
        /// <param name="id">消息id</param>
        /// <returns></returns>
        private T ReceiveByIdNoBlock(string id)
        {
            try
            {
                return TransferMessageToObject(msgQueue.ReceiveByCorrelationId(id, timeout));
            }
            catch (Exception e)
            {
                DealWithException(e);
            }

            return null;
        }

        #endregion

        #region 删除消息

        /// <summary>
        /// 根据消息队列的id删除消息
        /// </summary>
        /// <param name="id">消息队列id</param>
        public bool DeleteMsgById(string id)
        {
            bool result = true;
            try
            {
                msgQueue.MessageReadPropertyFilter.SetAll();
                msgQueue.ReceiveByCorrelationId(id);
            }
            catch (Exception e)
            {
                DealWithException(e);
                result = false;
            }

            return result;
        }

        #endregion

        #region 接收消息（接收后不删除消息）

        /// <summary>
        /// 返回但不移除所引用的队列中的第一条消息。在有可用消息前，该方法阻塞当前线程。
        /// </summary>
        /// <returns></returns>
        public T Peek()
        {
            return TransferMessageToObject(msgQueue.Peek());
        }

        /// <summary>
        /// 返回但不移除所引用的队列中的第一条消息，此方法不阻止当前线程的执行。
        /// </summary>
        /// <returns></returns>
        public T PeekNoBlock()
        {
            try
            {
                return TransferMessageToObject(msgQueue.Peek(timeout));
            }
            catch (Exception e)
            {
                DealWithException(e);
            }

            return null;
        }

        /// <summary>
        /// 查看其消息标识符匹配 id 参数的消息。
        /// </summary>
        /// <param name="id">消息id</param>
        /// <returns></returns>
        public T PeekById(string id)
        {
            try
            {
                return TransferMessageToObject(msgQueue.PeekByCorrelationId(id));
            }
            catch (Exception e)
            {
                DealWithException(e);
            }

            return null;
        }

        /// <summary>
        /// 查看其消息标识符匹配 id 参数的消息，此方法不阻止当前线程的执行。
        /// </summary>
        /// <param name="id">消息id</param>
        /// <returns></returns>
        private T PeekByIdNoBlock(string id)
        {
            try
            {
                return TransferMessageToObject(msgQueue.PeekByCorrelationId(id, timeout));
            }
            catch (Exception e)
            {
                DealWithException(e);
            }

            return null;
        }

        #endregion

        /// <summary>
        /// 判断消息队列是否为空
        /// </summary>
        public bool IsQueueEmpty
        {
            get
            {
                bool isQueueEmpty = false;
                try
                {
                    msgQueue.Peek(timeout);
                }
                catch (MessageQueueException e)
                {
                    if (e.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
                    {
                        isQueueEmpty = true;
                    }
                }

                return isQueueEmpty;
            }
        }

        /// <summary>
        /// 返回消息队列中的消息数目
        /// </summary>
        public int MessageCount
        {
            get
            {
                int msgCount;
                try
                {
                    msgCount = GetMessageCount(msgQueue.MachineName, msgQueue.FormatName);
                }
                catch
                {
                    int i = 0;
                    MessageEnumerator enumerator = msgQueue.GetMessageEnumerator2();
                    while (enumerator.MoveNext())
                    {
                        ++i;
                    }

                    msgCount = i;
                }

                return msgCount;
            }
        }

        /// <summary>
        /// 删除队列中包含的所有消息。
        /// </summary>
        public void Purge()
        {
            msgQueue.Purge();
        }

        /// <summary>
        /// 为队列中的所有消息创建枚举数对象。
        /// </summary>
        /// <returns></returns>
        private MessageEnumerator GetMessageEnumerator()
        {
            return msgQueue.GetMessageEnumerator2();
        }

        /// <summary>
        /// 返回位于队列中的所有消息。
        /// </summary>
        /// <returns></returns>
        public IList<T> GetAllMessages()
        {
            return GetTopMessages(Int32.MaxValue);
        }

        /// <summary>
        /// 返回位于队列中的前num条消息。
        /// </summary>
        /// <returns></returns>
        public IList<T> GetTopMessages(int num)
        {
            //解决 接收消息时未检索到属性Body bug
            msgQueue.MessageReadPropertyFilter.SetAll();
            msgQueue.MessageReadPropertyFilter.Body = true;

            lock (_loke)
            {
                IList<T> msgs = new List<T>();
                int i = 0;
                using (MessageEnumerator enumerator = msgQueue.GetMessageEnumerator2())
                {
                    while (enumerator.MoveNext())
                    {
                        Message msg = GetCurrentMsg(enumerator);
                        if (msg == null)
                        {
                            continue;
                        }
                        else
                        {
                            msgs.Add(TransferMessageToObject(msg));
                            i++;
                            if (i >= num) break;
                        }
                    }
                }

                return msgs;
            }
        }

        /// <summary>
        /// 移动队列中的前num条消息对目标消息队列
        /// 如果队列中的实际消息条数小于num，则移动实际条数
        /// </summary>
        /// <param name="target">目标消息队列</param>
        /// <param name="num">移动的消息条数</param>
        /// <returns></returns>
        public IList<T> MoveTo(MessageQueueEx<T> target, int num)
        {
            if (target == null)
            {
                throw new ArgumentNullException();
            }

            IList<T> msgs = new List<T>();
            for (int i = 0; i < num; i++)
            {
                if (!this.IsQueueEmpty)
                {
                    Message msg = msgQueue.Peek();
                    msgs.Add(TransferMessageToObject(msg));
                    target.Send(msg);
                    this.msgQueue.Receive();
                }
                else
                {
                    break;
                }
            }
            return msgs;
        }

        #region Private Method

        /// <summary>
        /// 发送消息到消息队列
        /// </summary>
        /// <param name="msg">Message消息</param>
        private void Send(Message msg)
        {
            msg.Recoverable = true;
            msgQueue.Send(msg);
        }

        /// <summary>
        /// 随机生成消息的CorrelationId
        /// </summary>
        /// <returns></returns>
        private string GetCorrelationId()
        {
            string guid = Guid.NewGuid().ToString();
            Random random = new Random(guid.GetHashCode());
            string randomstr = random.Next(99999).ToString().PadLeft(5, '1');
            string str = string.Format(@"{0}\{1}", guid, randomstr);
            return str;
        }

        /// <summary>
        /// 将Message转换成Object对象
        /// </summary>
        /// <param name="msg"></param>
        private T TransferMessageToObject(Message msg)
        {
            T obj = msg.Body as T;

            if (obj != null)
            {
                obj.Id = msg.CorrelationId;
            }

            return obj;
        }

        /// <summary>
        /// 处理异常
        /// </summary>
        /// <param name="e"></param>
        private void DealWithException(Exception e)
        {
            bool throwException = true;

            //检查是否是由于超时导致的异常。此类异常不向外抛出。
            if (e is MessageQueueException)
            {
                if (((MessageQueueException)e).MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
                {
                    throwException = false;
                }
            }
            //检查是否是由于找不到指定id的消息导致的异常。此类异常不向外抛出。
            else if (e is InvalidOperationException)
            {
                throwException = true;
                //string msg = msgQueue.FormatName;
                //BaseLog.Error(string.Format("根据id删除消息出错。消息队列FormatName{0}", msg));
            }

            if (throwException)
            {
                throw e;
            }
        }

        /// <summary>
        /// 得到游标的当前msg
        /// </summary>
        /// <param name="enumerator"></param>
        /// <returns></returns>
        private Message GetCurrentMsg(MessageEnumerator enumerator)
        {
            Message msg = null;
            try
            {
                msg = enumerator.Current;
            }
            catch (MessageQueueException ex)
            {
                if (ex.MessageQueueErrorCode != MessageQueueErrorCode.MessageAlreadyReceived)
                {
                    throw;
                }
            }

            return msg;
        }

        /// <summary>
        /// 得到消息队列中消息个数
        /// </summary>
        /// <param name="machineName">消息队列机器名</param>
        /// <param name="queueFormatName">消息队列FormatName</param>
        /// <returns>消息队列中消息总数</returns>
        private int GetMessageCount(string machineName, string queueFormatName)
        {
            //@"FormatName:DIRECT=TCP:192.168.1.22\Private$\gtwoutorder";
            MSMQ.MSMQManagement mgmt = new MSMQ.MSMQManagement();

            Object ss = (Object)machineName;
            Object pn = (Object)queueFormatName;
            Object f = (Object)null;

            mgmt.Init(ref ss, ref f, ref pn);

            return mgmt.MessageCount;
        }

        #endregion

        #region Dispose

        // 标记对象是否已被释放
        private bool disposed = false;

        ~MessageQueueEx()
        {
            Dispose(false);
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(true);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposed) return;

            if (disposing)
            {
                //在这里释放托管资源
                msgQueue.Dispose();
            }
            //在这里释放非托管资源
            //do some thing
            disposed = true;
        }

        #endregion
    }
}
